package com.atbeijing.D10;

import com.beust.jcommander.JCommander;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;

public class ClsTopology {
    public static void main(String[] args) {
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        //将命令行参数转为实体类
//        JdbcOption option = new JdbcOption();
//        JCommander jCommander = JCommander.newBuilder().addObject(option).build();
//        jCommander.parse(args);



        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.enableCheckpointing(10*1000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(120*1000L);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10*1000L));
        env.setStateBackend(new FsStateBackend("hdfs://hadoop202:8020/gmall/ck"));
//        System.setProperty("HADOOP_USER_NAME","wjg");


        env.execute("storm_to_flink_cls");

    }
}
